package com.kamistoat.meimeimall.meimeimallorder.listener;

import com.kamistoat.meimeimall.common.To.RabbitMQTo.OrderTo;
import com.kamistoat.meimeimall.common.To.cartTo.CartDeleteTo;
import com.kamistoat.meimeimall.common.To.cartTo.CartItemDeleteDetailTo;
import com.kamistoat.meimeimall.common.To.couponTo.CouponItemDetailTo;
import com.kamistoat.meimeimall.common.To.couponTo.CouponUpdateTo;
import com.kamistoat.meimeimall.common.To.seckillTo.SecKillProductCacheTo;
import com.kamistoat.meimeimall.common.To.wareTo.LockOrderItemWareTo;
import com.kamistoat.meimeimall.common.To.wareTo.LockOrderWareTo;
import com.kamistoat.meimeimall.common.constant.OrderConstant;
import com.kamistoat.meimeimall.common.constant.SecKillConstant;
import com.kamistoat.meimeimall.common.dubboInterface.CartDubboInterface;
import com.kamistoat.meimeimall.common.dubboInterface.CouponDubboInterface;
import com.kamistoat.meimeimall.common.dubboInterface.WareDubboInterface;
import com.kamistoat.meimeimall.common.exception.BizCodeEnum;
import com.kamistoat.meimeimall.common.exception.RRException;
import com.kamistoat.meimeimall.common.utils.R;
import com.kamistoat.meimeimall.meimeimallorder.dao.OrderDao;
import com.kamistoat.meimeimall.meimeimallorder.entity.OrderEntity;
import com.kamistoat.meimeimall.meimeimallorder.entity.OrderItemEntity;
import com.kamistoat.meimeimall.meimeimallorder.enums.OrderStatusEnum;
import com.kamistoat.meimeimall.meimeimallorder.localTCC.OrderServiceLocalTCC;
import com.kamistoat.meimeimall.meimeimallorder.service.OrderService;
import com.kamistoat.meimeimall.meimeimallorder.vo.OrderCreateVo;
import com.kamistoat.meimeimall.meimeimallorder.vo.OrderSubmitVo;
import com.rabbitmq.client.Channel;
import io.seata.spring.annotation.GlobalTransactional;
import org.apache.dubbo.config.annotation.Reference;
import org.redisson.api.RLock;
import org.redisson.api.RSemaphore;
import org.redisson.api.RedissonClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.math.BigDecimal;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * 当订单通过缓存预扣减后投递下单消息。
 * <p></p>
 * 是一个TCC全局事务
 */
@Service
public class OrderCreateUnpayListener {
    protected Logger logger = LoggerFactory.getLogger(OrderCreateUnpayListener.class);

    @Autowired
    RedissonClient redissonClient;
    @Autowired
    RabbitTemplate rabbitTemplate;
    @Autowired
    StringRedisTemplate stringRedisTemplate;

    @Autowired
    OrderService orderService;
    @Autowired
    OrderServiceLocalTCC orderServiceLocalTCC;

    @Autowired
    OrderDao orderDao;

    @Reference
    WareDubboInterface wareDubboInterface;
    @Reference
    CartDubboInterface cartDubboInterface;
    @Reference
    CouponDubboInterface couponDubboInterface;

    /**
     * <b>已弃用</b>
     * 监听到下单消息执行。
     * 注意TCC全局事务注解 {@link GlobalTransactional}
     *
     * @param message
     * @param orderSubmitVo
     * @param channel
     */
    @Deprecated
    @RabbitListener(queues = "order.create.unpay.queue")
    @GlobalTransactional
    public void createUnpayListener(Message message, OrderSubmitVo orderSubmitVo, Channel channel) throws IOException {
        logger.info("检测到来自用户" + orderSubmitVo.getUserId() + "的下单动作:");
        // 消息序号，用于全部成功后手动向MQ进行ACK确认
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        R createUnpayR = new R();
        // 两个用户在下单秒杀商品时回滚缓存数据的指针。
        boolean successSemaphore = false;
        boolean successUserCount = false;

        // 分布式锁。限制一个账户一个时间只能有一个登录实例执行下单
        RLock submitLock = redissonClient.getLock(OrderConstant.ORDER_SUBMIT_LOCK_PREFIX + orderSubmitVo.getUserId());
        try {
            /**
             * 1. 获取分布式锁
             */
            boolean submitTry = submitLock.tryLock(3, TimeUnit.SECONDS);
            // 没有获取到锁，说明存在多个登录实例同时提交订单，则直接抛出对应异常，由下方catch处理后继续抛出
            if (!submitTry) {
                createUnpayR = R.error(BizCodeEnum.SINGLE_USER_LIMITED.getCode(), BizCodeEnum.SINGLE_USER_LIMITED.getMessage());
                throw new RRException(BizCodeEnum.SINGLE_USER_LIMITED.getMessage(), BizCodeEnum.SINGLE_USER_LIMITED.getCode());
            }

            /**
             * 2. 如果是秒杀订单，扣除信号量锁，并使用LUA脚本原子修改累计购买数量.
             */
            if (orderSubmitVo.getIsSeckill()) {
                SecKillProductCacheTo cacheTo = orderSubmitVo.getSecKillProductCacheTo();

                RSemaphore semaphore = redissonClient.getSemaphore(SecKillConstant.SEC_PRODUCT_SEMAPHORE + orderSubmitVo.getSeckillRandomCode());
                boolean tryAcquire = semaphore.tryAcquire(orderSubmitVo.getSeckillNum(), 3, TimeUnit.SECONDS);
                // 信号量扣除失败则直接抛出异常
                if (!tryAcquire) {
                    throw new RRException(BizCodeEnum.SECKILL_VERIFY_NO_SEMAPHORE.getMessage(), BizCodeEnum.SECKILL_VERIFY_NO_SEMAPHORE.getCode());
                }
                successSemaphore = true;
                String userCountKey = SecKillConstant.SEC_USER_PRODUCT_COUNT + orderSubmitVo.getSeckillSessionId().toString() + "_" + orderSubmitVo.getUserId().toString();
                Long execute = stringRedisTemplate.execute(
                        new DefaultRedisScript<>(
                                "local alreadyBought = tonumber(redis.call('get', KEYS[1]));" +
                                        "if (alreadyBought + tonumber(ARGV[1]) > tonumber(ARGV[2])) then " +
                                        "return 1; " +
                                        "end; " +
                                        "redis.call('set', KEYS[1], alreadyBought + tonumber(ARGV[1]));" +
                                        "return 0;", Long.class)
                        , Collections.singletonList(userCountKey), orderSubmitVo.getSeckillNum(), cacheTo.getSecKillProductTo().getSeckillUserLimit());
                if (execute.equals(1L)) {
                    throw new RRException(BizCodeEnum.SECKILL_VERIFY_SELF_OVER.getMessage(), BizCodeEnum.SECKILL_VERIFY_SELF_OVER.getCode());
                }
                successUserCount = true;
            }

            /**
             * 3. 使用下单消息提交前固定的订单小项列表来构建完整的订单。
             *
             * 普通订单: 再去数据库根据购物车构建最终下单订单。
             * 秒杀订单: 由于秒杀商品不入购物车，所以直接根据秒杀信息确定商品即可。
             *
             * 此方法仅查询数据库，无修改.
             */
            OrderCreateVo orderCreateVo = orderService.createOrderBySubmit(orderSubmitVo);


            /**
             * 4. 价格验证. 验证前端传递价格和重新构建的是否一致。价格验证通过后，将开启修改数据库的全局事务
             */
            BigDecimal payAmount = orderCreateVo.getOrderEntity().getPayAmount();
            BigDecimal truePrice = orderSubmitVo.getShouldPayPrice();
            // 价格校验失败，同样直接抛出异常，由下方catch处理后继续抛出
            if (Math.abs(payAmount.subtract(truePrice).doubleValue()) >= 0.01) {
                createUnpayR = R.error(BizCodeEnum.ORDER_RECONFIRM_ERROR.getCode(), BizCodeEnum.ORDER_RECONFIRM_ERROR.getMessage());
                throw new RRException(BizCodeEnum.ORDER_RECONFIRM_ERROR.getMessage(), BizCodeEnum.ORDER_RECONFIRM_ERROR.getCode());
            }

            /**
             * 5. RPC锁定库存.
             * 分布式事务分支.
             */
            LockOrderWareTo lockOrderWareTo = new LockOrderWareTo();
            lockOrderWareTo.setOrderSn(orderCreateVo.getOrderEntity().getOrderSn());  // 订单的sn

            // 保存这次订单中所有商品小项的购买详情，ware模块将根据这些信息依次去锁定每一个商品的库存
            List<LockOrderItemWareTo> lockOrderItemWareToList = orderCreateVo.getOrderItemEntityList().stream().map(orderItemEntity -> {
                LockOrderItemWareTo lockOrderItemWareTo = new LockOrderItemWareTo();
                lockOrderItemWareTo.setSkuId(orderItemEntity.getSkuId());
                lockOrderItemWareTo.setCount(orderItemEntity.getSkuQuantity());
                lockOrderItemWareTo.setTitle(orderItemEntity.getSkuName());
                return lockOrderItemWareTo;
            }).collect(Collectors.toList());
            lockOrderWareTo.setLockList(lockOrderItemWareToList);  // 订单中所有商品小项的信息，放到wareSkuLockVo中

            // Dubbo调用，开启锁定库存的事务分支
            R orderLockStockR = wareDubboInterface.orderLockStock(null, lockOrderWareTo);

            if (orderLockStockR.getCode() != 0) {
                // RPC锁定库存失败，直接抛出异常，由下方catch处理后继续抛出
                createUnpayR = R.error(orderLockStockR.getCode(), orderLockStockR.getMsg());
                throw new RRException(orderLockStockR.getMsg(), BizCodeEnum.NO_STOCK_EXCEPTION.getCode());
            }
            /**
             * 6. 如果是普通订单，从购物车中删除当前订单包含的商品.
             * 分布式事务分支.
             */
            if (!orderSubmitVo.getIsSeckill()) {
                R deleteCartR = cartDubboInterface.deleteOrderItemInCart(null, lockOrderItemWareToList);
                if (deleteCartR.getCode() != 0) {
                    createUnpayR = R.error(deleteCartR.getCode(), deleteCartR.getMsg());
                    throw new RRException(deleteCartR.getMsg(), deleteCartR.getCode());
                }
            }

            /**
             * 7. 向消息队列投递消息: 超时未支付订单、超时库存锁定工单
             */
            OrderTo orderTo = new OrderTo();
            orderTo.setOrderId(orderCreateVo.getOrderEntity().getId());
            orderTo.setOrderSn(orderCreateVo.getOrderEntity().getOrderSn());
            orderTo.setMemberId(orderCreateVo.getOrderEntity().getMemberId());
            rabbitTemplate.convertAndSend("order-event-exchange", "order.delay", orderTo);
            rabbitTemplate.convertAndSend("stock-event-exchange", "stock.seckill.stock", lockOrderWareTo);


            /**
             * 8. 全部操作完成，则将订单保存入库.
             * 分布式事务分支。但本地调用，注意接口上额外的 {@link io.seata.rm.tcc.api.LocalTCC} 注解
             */
            orderServiceLocalTCC.saveUnpayOrder(null, orderCreateVo);

            createUnpayR = R.ok("可重定向至支付页面");
            rabbitTemplate.convertAndSend("order-event-exchange", "order.create.result", createUnpayR);

        } catch (Exception exception) {
            // 捕获上面可能抛出的所有异常。包括自定义异常和未知异常
            // 如果是秒杀商品，首先需要手动回滚缓存信息
            if (successSemaphore) {
                RSemaphore semaphore = redissonClient.getSemaphore(SecKillConstant.SEC_PRODUCT_SEMAPHORE + orderSubmitVo.getSeckillRandomCode());
                semaphore.release(orderSubmitVo.getSeckillNum());
            }
            if (successUserCount) {
                String userCountKey = SecKillConstant.SEC_USER_PRODUCT_COUNT + orderSubmitVo.getSeckillSessionId().toString() + "_" + orderSubmitVo.getUserId().toString();
                Long execute = stringRedisTemplate.execute(
                        new DefaultRedisScript<>(
                                "local alreadyBought = tonumber(redis.call('get', KEYS[1]));" +
                                        "redis.call('set', KEYS[1], alreadyBought - tonumber(ARGV[1]));" +
                                        "return 0;", Long.class)
                        , Collections.singletonList(userCountKey), orderSubmitVo.getSeckillNum());
            }

            // 根据异常构造R对象。自定义异常已经配置好了createUnpayR，只有未知异常还需要手动添加上异常信息
            if (!(exception instanceof RRException)) {
                createUnpayR = R.error(BizCodeEnum.UNKNOW_EXCEPTION.getCode(), exception.getMessage());
            }
            // 向下单结果队列发送消息
            rabbitTemplate.convertAndSend("order-event-exchange", "order.create.result", createUnpayR);
        } finally {
            // 无论成功与否，下单动作只进行一次，必然会确认消息
            channel.basicAck(deliveryTag, false);
            submitLock.unlock();
        }
    }


    /**
     * 消费MQ中的下单消息。
     * 注意，一旦缓存预扣除成功，消息进入MQ后，则一定保证消息的持久性。
     * 也就可以不再使用分布式事务，因为缓存预扣除成功代表了订单一定可以下成，那么全局事务中已经执行成功的事务没有必要在因异常导致的全局事务失败中自动回滚。
     * 因为之后这条失败消息一定会被重复消费，只需要做好这些操作的消费幂等即可。
     */
    @RabbitListener(queues = "order.create.unpay.queue")
    public void createUnpayListenerForAllType(Message message, OrderSubmitVo orderSubmitVo, Channel channel) throws IOException {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        Long userId = orderSubmitVo.getUserId();
        String orderSN = orderSubmitVo.getOrderSN();
        logger.info("检测到来自用户" + userId + "的下单动作:");

        /**
         * 抢占分布式锁，对于一个订单，只允许一个线程进行处理
         * 1. 防止消息被异常Rebalance时的重复消费
         * 2. 防止消息同时处于多个流转阶段，状态被重叠
         * 3. 抢占不到锁，直接拒绝ACK
         */
        RLock submitLock = redissonClient.getLock("order::cacheLock::{" + userId + "}::" + orderSN);
        try {
            // 非阻塞式抢占锁
            boolean submitTry = submitLock.tryLock();
            // 没有获取到锁，直接拒绝消息，重新投递后等待重新消费即可
            if (!submitTry) {
                channel.basicReject(deliveryTag, false);
            }

            // 抢占到锁

            /**
             * 从数据库查询是否存在相同SN订单。
             * 如果存在，1. 缓存预扣除阶段被补偿线程重复投递   2. 异常Rebalance后的重复消费
             */
            OrderEntity orderInDatabase = orderDao.selectByMemberIdAndOrderSnAndStatus(userId, orderSN, null);
            // 如果数据库中不存在相同SN的订单，则开始处理一条新的订单
            // 存在相同SN订单且状态为CREATING，说明是之前异常结束后的重复消费
            if (orderInDatabase == null || !orderInDatabase.getStatus().equals(OrderStatusEnum.CREATING.getCode())) {
                /**
                 * 根据预扣除时已经固定的商品列表，构建详细订单，包括
                 * 查询出订单中各个商品除价格外的详细信息、使用预扣除时固定的价格计算订单的真实价格
                 */
                OrderCreateVo orderCreateVo = orderService.createOrderBySubmit(orderSubmitVo);

                // 新处理的订单还需要先插入
                if(orderInDatabase == null) {
                    // 解析总订单，拆分成按照品牌区分的子订单，对总订单和子订单执行入库.
                    // 总订单与子订单的状态均为 CREATING.
                    // 不再需要分布式事务，只要入库本身是本地事务即可.
                    orderService.saveCreateOrder(orderCreateVo);
                }

                /**
                 * 开始执行各种动作的分支事务。根据锁竞争激烈程度的不同，以不同的思路执行
                 */

                // 删除购物车中的选中商品。
                // 不再需要分布式事务。
                CartDeleteTo cartDeleteTo = new CartDeleteTo();
                cartDeleteTo.setUserId(userId);
                cartDeleteTo.setOrderSN(orderSN);
                cartDeleteTo.setItemTos(
                        orderCreateVo.getOrderItemEntityList().stream().map((orderItemEntity -> {
                            CartItemDeleteDetailTo detail = new CartItemDeleteDetailTo();
                            detail.setSkuId(orderItemEntity.getSkuId());
                            detail.setCount(orderItemEntity.getSkuQuantity());
                            return detail;
                        })).collect(Collectors.toList()));
                R deletRes = cartDubboInterface.deleteOrderItem(cartDeleteTo);
                if (deletRes.getCode() != 0) {
                    throw new RRException(deletRes.getMsg(), deletRes.getCode());
                }

                // 修改用户积分
                // 由于锁竞争并不激烈，因此插入积分操作日志和修改积分可以同步执行，不必使用任务引擎. 当然也可以使用任务引擎
                // 不再需要分布式事务
                CouponUpdateTo couponUpdateTo = new CouponUpdateTo();
                couponUpdateTo.setUserId(userId);
                couponUpdateTo.setOrderSN(orderSN);
                couponUpdateTo.setUsedIntegration(300);
                couponUpdateTo.setUsedGrowth(300);
                couponUpdateTo.setItemTos(
                        orderCreateVo.getOrderItemEntityList().stream().map((orderItemEntity -> {
                            CouponItemDetailTo detail = new CouponItemDetailTo();
                            detail.setSkuId(orderItemEntity.getSkuId());
                            detail.setCount(orderItemEntity.getSkuQuantity());
                            detail.setGiftGrowth(orderItemEntity.getGiftGrowth());
                            detail.setGiftIntegration(orderItemEntity.getGiftIntegration());
                            return detail;
                        })).collect(Collectors.toList()));
                R updateRes = couponDubboInterface.updateOrderCoupon(couponUpdateTo);
                if (updateRes.getCode() != 0) {
                    throw new RRException(updateRes.getMsg(), updateRes.getCode());
                }

                // 锁定商品库存
                // 明显是一个锁竞争很激烈的动作，因此采用任务引擎的思路，此处只做订单-库存工单的落库，只要落库成功即认为库存扣减成功。
                // 之后的真实扣减由MQ去做。
                LockOrderWareTo lockOrderWareTo = new LockOrderWareTo();
                lockOrderWareTo.setUserId(userId);
                lockOrderWareTo.setOrderSn(orderSN);
                List<LockOrderItemWareTo> lockOrderItemWareToList = orderCreateVo.getOrderItemEntityList().stream().map(orderItemEntity -> {
                    LockOrderItemWareTo lockOrderItemWareTo = new LockOrderItemWareTo();
                    lockOrderItemWareTo.setSkuId(orderItemEntity.getSkuId());
                    lockOrderItemWareTo.setCount(orderItemEntity.getSkuQuantity());
                    lockOrderItemWareTo.setTitle(orderItemEntity.getSkuName());
                    return lockOrderItemWareTo;
                }).collect(Collectors.toList());
                lockOrderWareTo.setLockList(lockOrderItemWareToList);
                // 发送RPC请求，要求库存模块对订单-库存工单进行落库，只进行落库，不进行扣除
                // 同样，只需要是一个本地事务即可
                R logRes = wareDubboInterface.insertOrderWareOperation(lockOrderWareTo);
                if (logRes.getCode() != 0) {
                    throw new RRException(logRes.getMsg(), logRes.getCode());
                }

                // 下单动作完成后，更新订单和子订单的状态为待支付
                orderService.saveCreateDoneOrder(orderCreateVo);

                // 发送订单超时释放的定时消息
                OrderTo orderTo = new OrderTo();
                orderTo.setOrderId(orderCreateVo.getOrderEntity().getId());
                orderTo.setOrderSn(orderCreateVo.getOrderEntity().getOrderSn());
                orderTo.setMemberId(orderCreateVo.getOrderEntity().getMemberId());
                rabbitTemplate.convertAndSend("order-event-exchange", "order.delay", orderTo);
                rabbitTemplate.convertAndSend("stock-event-exchange", "stock.seckill.stock", lockOrderWareTo);

                // 告知创建结果
                R createUnpayR = R.ok("创建成功，可以执行支付");
                rabbitTemplate.convertAndSend("order-event-exchange", "order.create.result", createUnpayR);
            }
            // 确认这条消息
            channel.basicAck(deliveryTag, false);
        } catch (Exception exception) {
            R createUnpayR;
            if (!(exception instanceof RRException)) {
                createUnpayR = R.error(BizCodeEnum.UNKNOW_EXCEPTION.getCode(), "重试中..." + exception.getMessage());
            }else {
                createUnpayR = R.error(((RRException) exception).getCode(), "重试中..." + ((RRException) exception).getMsg());
            }
            // 告知创建结果
            rabbitTemplate.convertAndSend("order-event-exchange", "order.create.result", createUnpayR);
            // 拒绝这条消息
            channel.basicReject(deliveryTag, false);
        } finally {
            submitLock.unlock();
        }
    }
}
